#include <stddef.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <glib.h>

#include "octhreadpool.h"
#include "ocexception.h"
#include "ocatomic.h"

struct _oc_thread_pool_t
{
    guint32                 nthreads;
    //guint32                 nthreads_done;

    pthread_t               *thread;
    oc_usr_thread_func_t    usr_thread_func;
    sem_t                   *task_sem;
    sem_t                   *run_sem;
    gboolean                be_terminated;

    GAsyncQueue             **queue;
    guint32                 cur_queue_index;
};

typedef struct
{
    guint32                 nthreads;
    //guint32                 *nthreads_done;
    guint32                 thread_index;
    oc_usr_thread_func_t    usr_thread_func;
    sem_t                   **task_sem;
    sem_t                   *run_sem;
    gboolean                *be_terminated;
    GAsyncQueue             **queue;
}oc_thread_arg_t;

/*
* If the task is a general user task, its "data" will be a non-null
* pointer and its "type" will be filled with 1.
* If the task is a termination symbol meaning that all user tasks have been pushed
* into the distributed queues, its "data" will be a null pointer,
* and its "type" will be filled with 0.
* But in reference to the question that why we add a member variable
* "type" to identify if the task is a general user task or not,
* my answer is that it is not allowed that a null pointer is pushed into
* a "GAsyncQueue".
* When the thread has read a termination symbol, it will be closed.
* That a thread is closed means it won't go to work until
**/
typedef struct
{
    gpointer        *data;
    guint32         type;
}oc_task_t;


static void *oc_thread_func( gpointer arg );

void oc_thread_pool_init( oc_thread_pool_t *tpool,
                            oc_usr_thread_func_t usr_func )
{
    tpool->nthreads = sysconf ( _SC_NPROCESSORS_ONLN );
    //tpool->nthreads_done = 0;

    tpool->thread = g_new ( pthread_t, tpool->nthreads );
    tpool->usr_thread_func = usr_func;
    tpool->task_sem = g_new ( sem_t, tpool->nthreads );
    tpool->run_sem = g_new ( sem_t, tpool->nthreads );
    tpool->be_terminated = g_new ( gboolean, tpool->nthreads );

    tpool->queue = g_new ( GAsyncQueue *, tpool->nthreads );
    tpool->cur_queue_index = 0;

    pthread_attr_t attr;
    pthread_attr_init( & attr );
    pthread_attr_setdetachstate( & attr, PTHREAD_CREATE_DETACHED );
    pthread_attr_setscope( & attr, PTHREAD_SCOPE_SYSTEM );
    oc_thread_arg_t *arg;
    guint32 i, j;
    for ( i = 0; i != tpool->nthreads; ++i )
    {
        sem_init ( & tpool->task_sem[i], 0, 0 );
        sem_init ( & tpool->run_sem[i], 0, 0 );
        tpool->be_terminated[i] = TRUE;
        tpool->queue[i] = g_async_queue_new();

        arg = g_new( oc_thread_arg_t, 1 );
        arg->nthreads = tpool->nthreads;
        arg->nthreads_done = & tpool->nthreads_done;
        arg->thread_index = i;
        arg->usr_thread_func = tpool->usr_thread_func;
        arg->task_sem = g_new ( sem_t *, tpool->nthreads );
        for ( j = 0; j != tpool->nthreads; ++j )
        {
            arg->task_sem[j] = & tpool->task_sem[j];
        }
        arg->run_sem = & tpool->run_sem[i];
        arg->be_terminated = & tpool->be_terminated[i];
        arg->queue = tpool->queue;

        pthread_create( & tpool->thread[i], & attr, oc_thread_func, arg );
    }
    pthread_attr_destroy( & attr );
    /*
    guint32 count = total_num_tasks;
    while ( count > 0 )
    {
        for ( i = 0; i < tpool->nthreads; ++i )
        {
            ++tpool->ntasks_queue[i];
            if ( --count == 0 )
                break;
        }
    }*/
}

void oc_thread_pool_push (  oc_thread_pool_t *tpool,
                            gpointer usr_task )
{
    //oc_null_pointer_exception_check( tpool );
    //oc_null_pointer_exception_check( task );
    guint32 i = tpool->cur_queue_index;

    oc_task_t *task = g_new ( oc_task_t, 1 );
    task->data = usr_task;
    task->type = 1;

    g_async_queue_lock( tpool->queue[i] );
    g_async_queue_push_unlocked ( tpool->queue[i], task );
    sem_post ( & tpool->task_sem[i] );
    if ( tpool->be_terminated )
    {
        tpool->be_terminated[i] = FALSE;
        sem_post ( & tpool->run_sem[i] );
    }
    g_async_queue_unlock( tpool->queue[i] );

    tpool->cur_queue_index = ( tpool->cur_queue_index + 1 ) % tpool->nthreads;
}

static void *oc_thread_func( gpointer arg )
{
    oc_thread_arg_t *thread_arg = ( oc_thread_arg_t * ) arg;
    guint32 nthreads = thread_arg->nthreads;
    //guint32 *nthreads_done = thread_arg->nthreads_done;
    guint32 thread_index = thread_arg->thread_index;
    oc_usr_thread_func_t usr_thread_func = thread_arg->usr_thread_func;
    sem_t **task_sem = thread_arg->task_sem;
    sem_t *run_sem = thread_arg->run_sem;
    gboolean *be_terminated = thread_arg->be_terminated;
    GAsyncQueue **queue = thread_arg->queue;
    g_free ( thread_arg );
    oc_task_t *task = NULL;
    int sem_val = 0;

    while ( 1 )
    {
        sem_wait ( run_sem );
        if ( * be_terminated )
        {
            break;
        }
        while ( 1 )
        {
            /*
            * Wait until the user task pushed into the queue
            **/
            g_async_queue_lock ( queue[thread_index] );
            sem_getvalue ( task_sem[thread_index], & sem_val );
            if ( sem_val > 0 )
            {
                if ( ( task = g_async_queue_try_pop_unlocked ( queue[thread_index] ) ) )
                {
                    sem_try_wait( task_sem[thread_index] );
                    g_async_queue_unlock ( queue[thread_index] );
                    if ( task->type == 1 )
                    {
                        usr_thread_func( task->data );
                        g_free ( task );
                    }
                    else
                    {
                        g_free ( task );
                        /*
                        * Steal the task in other queues
                        **/
                    }
                }
                else
                {
                    g_async_queue_unlock ( queue[thread_index] );
                    //Handle the exception
                }
            }
            else
            {
                g_async_queue_unlock ( queue[thread_index] );
                /*
                * Steal the task in other queues
                **/
                guint32 i;
                for (   i = ( thread_index + 1 ) % nthreads;
                        i != thread_index;
                        i = ( i + 1 ) % nthreads )
                {
                    g_async_queue_lock ( queue[i] );
                    sem_getvalue ( task_sem[i], & sem_val );
                    if ( sem_val > 0 )
                    {
                        if ( ( task = g_async_queue_try_pop_unlocked ( queue[i] ) ) )
                        {
                            if ( task->type == 1 )
                            {
                                sem_try_wait( task_sem[i] );
                                g_async_queue_unlock ( queue[i] );
                                usr_thread_func( task->data );
                                g_free ( task );
                                break;
                            }
                            else
                            {
                                g_async_queue_push_unlocked ( queue[i], task );
                                g_async_queue_unlock ( queue[i] );
                            }
                        }
                        else
                        {
                            g_async_queue_unlock ( queue[i] );
                            //Handle the exception
                        }
                    }
                    g_async_queue_unlock ( queue[i] );
                }
            }

        }//while

    }//while

    return NULL;
}

void oc_thread_pool_barrier( oc_thread_pool_t *tpool )
{
    guint32 i;
    oc_task_t *task = NULL;
    for ( i = 0; i < tpool->nthreads; ++i )
    {
        task = g_new ( oc_task_t, 1 );
        task->data = NULL;
        task->type = 0;
        g_async_queue_lock ( tpool->queue[i] );
        g_async_queue_push_unlocked ( tpool->queue[i], task );
        sem_post ( & tpool->task_sem[i] );
        g_async_queue_unlock ( tpool->queue[i] );
    }
}

void oc_thread_pool_destroy( oc_thread_pool_t *tpool )
{
}
